/*
 * Copyright (c) 2014, marco.tamburelli@gmail.com
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met: 
 * 
 * 1. Redistributions of source code must retain the above copyright notice, this
 *    list of conditions and the following disclaimer. 
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution. 
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
 * ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
package org.omorphdb.core.index.hash;

import static org.omorphdb.core.value.ValueConstants._BYTE;
import static org.omorphdb.core.value.ValueConstants._INT;
import static org.omorphdb.core.value.ValueConstants._SHORT;

import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Date;

import org.omorphdb.core.exception.CorruptedStorageException;
import org.omorphdb.core.io.Constants;
import org.omorphdb.core.value.ValueType;

/**
 * Requires to be initialized with a type which will decide how to convert
 * element of such type to a hash value.
 * 
 * @author Marco Tamburelli
 */
public class HashFunctionFactory<T>
{
	private Converter<T> converter;
	private final ValueType keyType;

	interface Converter<T>
	{
		byte[] convert(T key);
	}

	/**
	 * Constructor.
	 * 
	 * @param keyType The chosen type of expected key value to process.
	 */
	@SuppressWarnings("unchecked")
	public HashFunctionFactory(ValueType keyType)
	{
		this.keyType = keyType;

		switch (this.keyType)
		{
		case BYTE:
			this.converter = (Converter<T>) new Converter<Byte>()
			{
				@Override
				public byte[] convert(Byte value)
				{
					return new byte[] { value };
				}
			};
			break;

		case CHAR:
			this.converter = (Converter<T>) new Converter<Character>()
			{
				@Override
				public byte[] convert(Character value)
				{

					byte[] array = new byte[Constants.SIZEOF_CHAR];
					ByteBuffer.wrap(array).putChar(value);

					return array;
				}
			};
			break;

		case SHORT:
			this.converter = (Converter<T>) new Converter<Short>()
			{
				@Override
				public byte[] convert(Short value)
				{
					byte[] array = new byte[Constants.SIZEOF_SHORT];
					ByteBuffer.wrap(array).putShort(value);

					return array;
				}
			};
			break;

		case INT:
			this.converter = (Converter<T>) new Converter<Integer>()
			{
				@Override
				public byte[] convert(Integer value)
				{
					byte[] array = new byte[Constants.SIZEOF_INT];
					ByteBuffer.wrap(array).putInt(value);

					return array;
				}
			};
			break;

		case FLOAT:
			this.converter = (Converter<T>) new Converter<Float>()
			{
				@Override
				public byte[] convert(Float value)
				{
					byte[] array = new byte[Constants.SIZEOF_FLOAT];
					ByteBuffer.wrap(array).putFloat(value);

					return array;
				}
			};
			break;

		case LONG:
			this.converter = (Converter<T>) new Converter<Long>()
			{
				@Override
				public byte[] convert(Long value)
				{

					byte[] array = new byte[Constants.SIZEOF_LONG];
					ByteBuffer.wrap(array).putLong(value);

					return array;
				}
			};
			break;

		case DATE:
			this.converter = (Converter<T>) new Converter<Date>()
			{
				@Override
				public byte[] convert(Date value)
				{
					byte[] array = new byte[Constants.SIZEOF_LONG];
					ByteBuffer.wrap(array).putLong(value.getTime());

					return array;
				}
			};
			break;

		case DOUBLE:
			this.converter = (Converter<T>) new Converter<Double>()
			{
				@Override
				public byte[] convert(Double value)
				{
					byte[] array = new byte[Constants.SIZEOF_DOUBLE];
					ByteBuffer.wrap(array).putDouble(value);

					return array;
				}
			};
			break;

		case STRING:
			this.converter = (Converter<T>) new Converter<String>()
			{
				@Override
				public byte[] convert(String value)
				{
					return value.getBytes();
				}
			};
			break;
		default:
			/*
			 * It shouldn't be possible since other values doesn't allow a hash
			 * value
			 */
			break;
		}
	}

	/**
	 * The class implementing the method which converts a key value to a byte
	 * array.
	 * 
	 * @return
	 */
	Converter<T> converter()
	{
		return this.converter;
	}

	/**
	 * Returns a {@link HashFunctionFactory} implementation implementing the
	 * algorithm corresponding to the provided byte.
	 * 
	 * @param hashAlgorithm
	 * @return
	 */
	public HashFunction<T> getHashFunction(byte hashAlgorithm)
	{
		switch (hashAlgorithm)
		{
		case _BYTE:
			return new HashFunction<T>()
			{
				@Override
				public int calculate(T key)
				{
					byte[] array = HashFunctionFactory.this.converter.convert(key);
					return HashFunctionFactory.toByte(array);
				}

				@Override
				public int maxValue()
				{
					return Byte.MAX_VALUE;
				}
			};
		case _SHORT:
			return new HashFunction<T>()
			{
				@Override
				public int calculate(T key)
				{
					byte[] array = HashFunctionFactory.this.converter.convert(key);
					return HashFunctionFactory.toShort(array);
				}

				@Override
				public int maxValue()
				{
					return Short.MAX_VALUE;
				}
			};
		case _INT:
			return new HashFunction<T>()
			{
				@Override
				public int calculate(T key)
				{
					byte[] array = HashFunctionFactory.this.converter.convert(key);
					return HashFunctionFactory.toInt(array);
				}

				@Override
				public int maxValue()
				{
					return Integer.MAX_VALUE;
				}
			};
		default:
			throw new CorruptedStorageException("Corrupted index");
		}
	}

	ValueType keyType()
	{
		return this.keyType;
	}

	static byte toByte(byte[] array)
	{
		ByteBuffer buf = ByteBuffer.wrap(array);

		byte hash = -1;
		while (buf.position() < buf.limit())
		{
			byte s1 = hash;
			s1 <<= 1;
			hash >>>= -1;
			hash |= s1;
			hash ^= buf.get();
		}

		if (hash < 0) hash ^= -1;

		return hash;
	}

	static short toShort(byte[] array)
	{
		int l = array.length + Constants.SIZEOF_SHORT - (array.length % Constants.SIZEOF_SHORT);
		byte[] a = Arrays.copyOf(array, l);

		ByteBuffer buf = ByteBuffer.wrap(a);

		short hash = -1;
		while (buf.position() < buf.limit())
		{
			short s1 = hash;
			s1 <<= 1;
			hash >>>= -1;
			hash |= s1;
			hash ^= buf.getShort();
		}

		if (hash < 0) hash ^= -1;

		return hash;
	}

	static int toInt(byte[] array)
	{
		int l = array.length + Constants.SIZEOF_INT - (array.length % Constants.SIZEOF_INT);
		byte[] a = Arrays.copyOf(array, l);

		ByteBuffer buf = ByteBuffer.wrap(a);

		int hash = -1;
		while (buf.position() < buf.limit())
		{
			hash = (hash << 1) | (hash >>> -1);
			hash ^= buf.getInt();
		}

		if (hash < 0) hash ^= -1;

		return hash;
	}

	@Deprecated
	static long toLong(byte[] array)
	{
		int l = array.length + Constants.SIZEOF_LONG - (array.length % Constants.SIZEOF_LONG);
		byte[] a = Arrays.copyOf(array, l);

		ByteBuffer buf = ByteBuffer.wrap(a);

		long hash = -1;
		while (buf.position() < buf.limit())
		{
			hash = (hash << 1) | (hash >>> -1);
			hash ^= buf.getLong();
		}

		if (hash < 0) hash ^= -1;

		return hash;
	}
}
